package cn.edu.xmu.datamining.tangzk.mralgos.pagerank;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.FileUtil;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.Tool;

import cn.edu.xmu.datamining.tangzk.mralgos.AlgorithmDriver;

public class PageRankDriver extends Configured implements Tool {

	public static final String USER_DEFINED_GROUP = "UserDefined";
	public static final String SUM_PAGERANK_COUNTER = "SumPageRank";
	public static final long PAGERANK_BASE = 10000;
	public static final String BETA_NAME = "pagerank.beta";
	public static final String TOTAL_NODES = "Total Nodes";
	public static final String SUM_PAGERANK_DIFF = "Sum PageRank Diff";

	@Override
	public int run(String[] args) throws Exception {
		if (args.length < 2) {
			throw new Exception(
					"Usage: <prog> <inDir> <outDir> [beta=1.0] [maxIteration=10]");
		}
		int exitCode = 0;

		Path inputDir = new Path(args[0]);
		String baseOutputPath = args[1].trim();

		float beta = 1.0f;
		int maxIteration = 10;
		if (args.length >= 3) {
			beta = Float.parseFloat(args[2]);
		}
		if (args.length >= 4) {
			maxIteration = Integer.parseInt(args[3]);
		}

		Configuration conf = this.getConf();
		conf.setFloat(PageRankDriver.BETA_NAME, beta);
		FileSystem fs = FileSystem.get(conf);
		FileUtil.copy(fs, inputDir, fs, new Path(baseOutputPath + "-iter0"),
				false, conf);

		Job job = null;
		int iteration = 0;
		double err = 1e-3;
		double sumPageRankDiff = 1.0f;

		while (sumPageRankDiff > err && iteration < maxIteration
				&& exitCode == 0) {
			job = new Job(conf, PageRankDriver.class.getSimpleName() + "-iter"
					+ iteration);

			job.setMapperClass(PageRankMapper.class);
			job.setReducerClass(PageRankReducer.class);
			job.setPartitionerClass(PageRankPartitioner.class);
			job.setGroupingComparatorClass(PageRankGrouper.class);

			job.setMapOutputKeyClass(PageRankNodeKeyWritable.class);
			job.setMapOutputValueClass(PageRankNodeValueWritable.class);
			job.setOutputKeyClass(IntWritable.class);
			job.setOutputValueClass(Text.class);

			FileInputFormat.addInputPath(job, new Path(baseOutputPath + "-iter"
					+ iteration));
			FileOutputFormat.setOutputPath(job, new Path(baseOutputPath
					+ "-iter" + (iteration + 1)));

			exitCode = AlgorithmDriver.runJob(job, false);

			sumPageRankDiff = job
					.getCounters()
					.findCounter(PageRankDriver.USER_DEFINED_GROUP,
							PageRankDriver.SUM_PAGERANK_DIFF).getValue()
					/ (PageRankDriver.PAGERANK_BASE * 1.0f);
			iteration++;
		}
		return exitCode;
	}

	public static void main(String[] args) throws Exception {
		AlgorithmDriver.toolRun(new PageRankDriver(), args);
	}

}
